// 版权所有2016 etcd作者
// 
// 根据Apache许可证2.0版（以下简称“许可证”）获得许可；
// 除非遵守许可证，否则不得使用此文件。
// 您可以通过
// 
// www.apache.org/licenses/LICENSE-2.0 
// 
// 除非适用法律要求或书面同意，否则根据许可证分发的软件
// 按“原样”分发，
// 无任何明示或暗示的保证或条件。
// 请参阅许可证，了解管理许可的特定语言和
// 许可证下的限制。

package clientv3

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
	"go.etcd.io/etcd/api/v3/mvccpb"
	v3rpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"

	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

const (
	EventTypeDelete = mvccpb.DELETE
	EventTypePut    = mvccpb.PUT

	closeSendErrTimeout = 250 * time.Millisecond
)

type Event mvccpb.Event

type WatchChan <-chan WatchResponse

type Watcher interface {
	// 按键或前缀观看手表。观看的事件将通过返回通道返回
	// 。如果压缩了等待通过
	// 监视发送的修订，则服务器将取消该监视，
	// 客户端将发布压缩的错误监视响应，通道将关闭。
	// 如果请求的修订为0或未指定，返回的通道将返回
	// 在服务器收到监视请求后发生的监视事件。
	// 如果上下文“ctx”被取消或超时，返回的“WatchChan”被关闭，
	// 和来自此关闭通道的“WatchResponse”具有零事件和零“Err（）”。
	// 必须取消上下文“ctx”，一旦不再使用观察程序，
	// 才能释放相关资源。
	// 
	// 如果上下文为“context.Background/TODO”，则返回的“WatchChan”将
	// 在事件触发之前不会关闭和阻止，除非服务器
	// 返回不可恢复的错误（例如errcompressed）。
	// 例如，当使用“WithRequiredReader”传递上下文，并且
	// 连接的服务器没有前导（例如，由于网络分区），
	// 将返回错误“etcdserver:无前导”（ErrNoLeader），
	// 然后使用非nil“Err（）”关闭“WatchChan”。
	// 为了防止监视流卡在分区节点中，
	// 请确保使用“WithRequiredReader”包装上下文。
	// 
	// 否则，只要上下文未被取消或超时，
	// watch将永远重试其他可恢复的错误，直到重新连接。
	// 
	// TODO:在最后一条“WatchResponse”消息中显式设置上下文错误并关闭通道？
	// 当前，客户端上下文被永远不会关闭的“valCtx”覆盖。
	// TODO（v3.4）：配置监视重试策略，限制最大重试次数
	// （请参阅https:
	Watch(ctx context.Context, key string, opts ...OpOption) WatchChan

	// RequestProgress请求在所有监视通道中发送进度通知响应。
	RequestProgress(ctx context.Context) error

	// Close关闭监视程序并取消所有监视请求。
	Close() error
}

type WatchResponse struct {
	Header pb.ResponseHeader
	Events []*Event

	// CompactRevision是监视程序可能收到的最小版本。
	CompactRevision int64

	// Cancelled用于指示监视失败。
	// 如果监视失败且流即将关闭，则在通道关闭之前，
	// 通道发送一个最终响应，该响应已取消设置为true，且错误为非nil（）.
	Canceled bool

	// Created用于指示观察者的创建。
	Created bool

	closeErr error

	// cancelReason是取消观察的原因
	cancelReason string
}

// IsCreate在事件通知密钥是新创建的情况下返回true。
func (e *Event) IsCreate() bool {
	return e.Type == EventTypePut && e.Kv.CreateRevision == e.Kv.ModRevision
}

// IsModify在事件通知在现有密钥上放置了新值时返回true。
func (e *Event) IsModify() bool {
	return e.Type == EventTypePut && e.Kv.CreateRevision != e.Kv.ModRevision
}

// Err是此WatchResponse包含错误时的错误值。如果WatchResponse是进度通知，则IsProgressNotify返回true。
func (wr *WatchResponse) Err() error {
	switch {
	case wr.closeErr != nil:
		return v3rpc.Error(wr.closeErr)
	case wr.CompactRevision != 0:
		return v3rpc.ErrCompacted
	case wr.Canceled:
		if len(wr.cancelReason) != 0 {
			return v3rpc.Error(status.Error(codes.FailedPrecondition, wr.cancelReason))
		}
		return v3rpc.ErrFutureRev
	}
	return nil
}

func (wr *WatchResponse) IsProgressNotify() bool {
	return len(wr.Events) == 0 && !wr.Canceled && !wr.Created && wr.CompactRevision == 0 && wr.Header.Revision != 0
}

// watcher实现watcher接口
type watcher struct {
	remote   pb.WatchClient
	callOpts []grpc.CallOption

	// mu保护grpc流映射
	mu sync.Mutex

	// 流包含所有活动的grpc流由ctx值键入。
	streams map[string]*watchGrpcStream
	lg      *zap.Logger
}

// watchGrpcStream跟踪附加到单个grpc流的所有监视资源。
type watchGrpcStream struct {
	owner    *watcher
	remote   pb.WatchClient
	callOpts []grpc.CallOption

	// ctx控制内部远程。watch requests 
	ctx context.Context
	// ctxKey是查找此流的上下文时使用的键
	ctxKey string
	cancel context.CancelFunc

	// 子流保存此grpc流上的所有活动监视程序EFG
	substreams map[int64]*watcherStream
	resuming []*watcherStream

	// reqc从watch（）发送监视请求到主goroutine 
	reqc chan watchStreamRequest
	// respc从监视客户端接收数据
	respc chan *pb.WatchResponse
	// donec关闭广播关闭
	donec chan struct{}
	// errc将错误从grpc Recv传输到监视流重新连接逻辑
	errc chan error
	// closingc获取关闭观察者的观察者流
	closingc chan *watcherStream
	// wg在所有子流Goroutine退出时完成
	wg sync.WaitGroup

	// resumec关闭以表示所有子流应开始恢复
	resumec chan struct{}
	// closeErr是关闭观察流
	closeErr error

	lg *zap.Logger
}

// watchStreamRequest是一个错误支持的监视请求操作类型的联合
type watchStreamRequest interface {
	toPB() *pb.WatchRequest
}

// 订阅服务器发出监视请求以启动新的监视程序
type watchRequest struct {
	ctx context.Context
	key string
	end string
	rev int64

	// 如果此字段为true，则发送创建的通知事件
	createdNotify bool
	// progressNotify用于进度更新
	progressNotify bool
	// 默认情况下应禁用分段
	// 如果为true，则在总数超过
	// /“--最大请求字节数”时拆分监视事件flag value+512字节
	fragment bool

	// filters是要过滤掉的事件列表
	filters []pb.WatchCreateRequest_FilterType
	// 在事件发生之前获取上一个键值对
	prevKV bool
	// retc在观察者建立后接收chan WatchResponse 
	retc chan chan WatchResponse
}

// progressRequest由订阅者发出请求监视进度
type progressRequest struct {
}

// watcherStream表示注册的观察者
type watcherStream struct {
	// initReq是启动此请求的请求
	initReq watchRequest

	// outc将观察响应发布到订阅者
	outc chan WatchResponse
	// recvc在发布
	recvc chan *WatchResponse
	// donec在观察流goroutine时关闭停止。
	donec chan struct{}
	// 当应计划关闭流时，关闭设置为true。
	closing bool
	// id是grpc流上注册的监视id 
	id int64

	// buf保存从etcd接收到但客户端尚未使用的所有事件
	buf []*WatchResponse
}

func NewWatcher(c *Client) Watcher {
	return NewWatchFromWatchClient(pb.NewWatchClient(c.conn), c)
}

func NewWatchFromWatchClient(wc pb.WatchClient, c *Client) Watcher {
	w := &watcher{
		remote:  wc,
		streams: make(map[string]*watchGrpcStream),
	}
	if c != nil {
		w.callOpts = c.callOpts
		w.lg = c.lg
	}
	return w
}

// 从不使用值关闭
var valCtxCh = make(chan struct{})
var zeroTime = time.Unix(0, 0)

// ctx；从未完成
type valCtx struct{ context.Context }

func (vc *valCtx) Deadline() (time.Time, bool) { return zeroTime, false }
func (vc *valCtx) Done() <-chan struct{}       { return valCtxCh }
func (vc *valCtx) Err() error                  { return nil }

func (w *watcher) newWatcherGrpcStream(inctx context.Context) *watchGrpcStream {
	ctx, cancel := context.WithCancel(&valCtx{inctx})
	wgs := &watchGrpcStream{
		owner:      w,
		remote:     w.remote,
		callOpts:   w.callOpts,
		ctx:        ctx,
		ctxKey:     streamKeyFromCtx(inctx),
		cancel:     cancel,
		substreams: make(map[int64]*watcherStream),
		respc:      make(chan *pb.WatchResponse),
		reqc:       make(chan watchStreamRequest),
		donec:      make(chan struct{}),
		errc:       make(chan error, 1),
		closingc:   make(chan *watcherStream),
		resumec:    make(chan struct{}),
		lg:         w.lg,
	}
	go wgs.run()
	return wgs
}

// Watch发布一个Watch请求以运行（），并等待新的watcher通道
func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan {
	ow := opWatch(key, opts...)

	var filters []pb.WatchCreateRequest_FilterType
	if ow.filterPut {
		filters = append(filters, pb.WatchCreateRequest_NOPUT)
	}
	if ow.filterDelete {
		filters = append(filters, pb.WatchCreateRequest_NODELETE)
	}

	wr := &watchRequest{
		ctx:            ctx,
		createdNotify:  ow.createdNotify,
		key:            string(ow.key),
		end:            string(ow.end),
		rev:            ow.rev,
		progressNotify: ow.progressNotify,
		fragment:       ow.fragment,
		filters:        filters,
		prevKV:         ow.prevKV,
		retc:           make(chan chan WatchResponse, 1),
	}

	ok := false
	ctxKey := streamKeyFromCtx(ctx)

	var closeCh chan WatchResponse
	for {
		// 查找或分配适当的grpc Watch stream 
		w.mu.Lock()
		if w.streams == nil {
			// 关闭
			w.mu.Unlock()
			ch := make(chan WatchResponse)
			close(ch)
			return ch
		}
		wgs := w.streams[ctxKey]
		if wgs == nil {
			wgs = w.newWatcherGrpcStream(ctx)
			w.streams[ctxKey] = wgs
		}
		donec := wgs.donec
		reqc := wgs.reqc
		w.mu.Unlock()

		// 无法创建通道；返回封闭通道
		if closeCh == nil {
			closeCh = make(chan WatchResponse, 1)
		}

		// 提交请求
		select {
		case reqc <- wr:
			ok = true
		case <-wr.ctx.Done():
			ok = false
		case <-donec:
			ok = false
			if wgs.closeErr != nil {
				closeCh <- WatchResponse{Canceled: true, closeErr: wgs.closeErr}
				break
			}
			// 重试；可能已从无ctxs 
			continue
		}

		// 接收通道
		if ok {
			select {
			case ret := <-wr.retc:
				return ret
			case <-ctx.Done():
			case <-donec:
				if wgs.closeErr != nil {
					closeCh <- WatchResponse{Canceled: true, closeErr: wgs.closeErr}
					break
				}
				// 删除流，重试；可能已经从没有CTXS ABCDEG 
				continue
			}
		}
		break
	}

	close(closeCh)
	return closeCh
}

func (w *watcher) Close() (err error) {
	w.mu.Lock()
	streams := w.streams
	w.streams = nil
	w.mu.Unlock()
	for _, wgs := range streams {
		if werr := wgs.close(); werr != nil {
			err = werr
		}
	}
	if err == context.Canceled {
		err = nil
	}
	return err
}

// RequestProgress请求在所有监视通道中发送进度通知响应。
func (w *watcher) RequestProgress(ctx context.Context) (err error) {
	ctxKey := streamKeyFromCtx(ctx)

	w.mu.Lock()
	if w.streams == nil {
		w.mu.Unlock()
		return fmt.Errorf("no stream found for context")
	}
	wgs := w.streams[ctxKey]
	if wgs == nil {
		wgs = w.newWatcherGrpcStream(ctx)
		w.streams[ctxKey] = wgs
	}
	donec := wgs.donec
	reqc := wgs.reqc
	w.mu.Unlock()

	pr := &progressRequest{}

	select {
	case reqc <- pr:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-donec:
		if wgs.closeErr != nil {
			return wgs.closeErr
		}
		// 重试；可能已从无ctxs 
		return w.RequestProgress(ctx)
	}
}

func (w *watchGrpcStream) close() (err error) {
	w.cancel()
	<-w.donec
	select {
	case err = <-w.errc:
	default:
	}
	return toErr(w.ctx, err)
}

func (w *watcher) closeStream(wgs *watchGrpcStream) {
	w.mu.Lock()
	close(wgs.donec)
	wgs.cancel()
	if w.streams != nil {
		delete(w.streams, wgs.ctxKey)
	}
	w.mu.Unlock()
}

func (w *watchGrpcStream) addSubstream(resp *pb.WatchResponse, ws *watcherStream) {
	// 删除流，检查手表ID是否向后兼容（<=v3.3）
	if resp.WatchId == -1 || (resp.Canceled && resp.CancelReason != "") {
		w.closeErr = v3rpc.Error(errors.New(resp.CancelReason))
		// 失败；无通道
		close(ws.recvc)
		return
	}
	ws.id = resp.WatchId
	w.substreams[ws.id] = ws
}

func (w *watchGrpcStream) sendCloseSubstream(ws *watcherStream, resp *WatchResponse) {
	select {
	case ws.outc <- *resp:
	case <-ws.initReq.ctx.Done():
	case <-time.After(closeSendErrTimeout):
	}
	close(ws.outc)
}

func (w *watchGrpcStream) closeSubstream(ws *watcherStream) {
	// 在从未建立流的情况下发送通道响应
	select {
	case ws.initReq.retc <- ws.outc:
	default:
	}
	// 关闭订阅者的通道
	if closeErr := w.closeErr; closeErr != nil && ws.initReq.ctx.Err() == nil {
		go w.sendCloseSubstream(ws, &WatchResponse{Canceled: true, closeErr: w.closeErr})
	} else if ws.outc != nil {
		close(ws.outc)
	}
	if ws.id != -1 {
		delete(w.substreams, ws.id)
		return
	}
	for i := range w.resuming {
		if w.resuming[i] == ws {
			w.resuming[i] = nil
			return
		}
	}
}

// run是用于管理观察者客户端
func (w *watchGrpcStream) run() {
	var wc pb.Watch_WatchClient
	var closeErr error

	// 标记为关闭但GOROUTE仍在运行的子流的根；需要
	// 避免在grpc流上双重关闭recvc拆卸
	closing := make(map[*watcherStream]struct{})

	defer func() {
		w.closeErr = closeErr
		// 关闭子流并恢复子流
		for _, ws := range w.substreams {
			if _, ok := closing[ws]; !ok {
				close(ws.recvc)
				closing[ws] = struct{}{}
			}
		}
		for _, ws := range w.resuming {
			if _, ok := closing[ws]; ws != nil && !ok {
				close(ws.recvc)
				closing[ws] = struct{}{}
			}
		}
		w.joinSubstreams()
		for range closing {
			w.closeSubstream(<-w.closingc)
		}
		w.wg.Wait()
		w.owner.closeStream(w)
	}()

	// 使用etcd grpc服务器
	if wc, closeErr = w.newWatchClient(); closeErr != nil {
		return
	}

	cancelSet := make(map[int64]struct{})

	var cur *pb.WatchResponse
	for {
		select {
		// Watch（）请求
		case req := <-w.reqc:
			switch wreq := req.(type) {
			case *watchRequest:
				outc := make(chan WatchResponse, 1)
				// TODO:传递自定义监视ID？
				ws := &watcherStream{
					initReq: *wreq,
					id:      -1,
					outc:    outc,
					// 取消缓冲，因此恢复不会导致重复事件
					recvc: make(chan *WatchResponse),
				}

				ws.donec = make(chan struct{})
				w.wg.Add(1)
				go w.serveSubstream(ws, w.resumec)

				// 排队创建观察者/恢复
				w.resuming = append(w.resuming, ws)
				if len(w.resuming) == 1 {
					// 恢复队列的负责人，可以注册新观察者
					if err := wc.Send(ws.initReq.toPB()); err != nil {
						w.lg.Debug("error when sending request", zap.Error(err))
					}
				}
			case *progressRequest:
				if err := wc.Send(wreq.toPB()); err != nil {
					w.lg.Debug("error when sending request", zap.Error(err))
				}
			}

		// 来自观察客户端的新事件
		case pbresp := <-w.respc:
			if cur == nil || pbresp.Created || pbresp.Canceled {
				cur = pbresp
			} else if cur != nil && cur.WatchId == pbresp.WatchId {
				// 合并新事件
				cur.Events = append(cur.Events, pbresp.Events...)
				// 更新“片段”字段；最后一个带有“片段”的响应==false 
				cur.Fragment = pbresp.Fragment
			}

			switch {
			case pbresp.Created:
				// 对队列头创建的响应
				if len(w.resuming) != 0 {
					if ws := w.resuming[0]; ws != nil {
						w.addSubstream(pbresp, ws)
						w.dispatchEvent(pbresp)
						w.resuming[0] = nil
					}
				}

				if ws := w.nextResume(); ws != nil {
					if err := wc.Send(ws.initReq.toPB()); err != nil {
						w.lg.Debug("error when sending request", zap.Error(err))
					}
				}

				// 为下一次迭代重置
				cur = nil

			case pbresp.Canceled && pbresp.CompactRevision == 0:
				delete(cancelSet, pbresp.WatchId)
				if ws, ok := w.substreams[pbresp.WatchId]; ok {
					// 向流GOROUTE发送信号以更新closingc 
					close(ws.recvc)
					closing[ws] = struct{}{}
				}

				// 为下一次迭代重置
				cur = nil

			case cur.Fragment:
				// 监视响应事件仍为碎片
				// 继续获取下一个碎片事件到达
				continue

			default:
				// 分派到相应的监视流
				ok := w.dispatchEvent(cur)

				// 为下一次迭代重置
				cur = nil

				if ok {
					break
				}

				// 对意外监视id的监视响应；取消id 
				if _, ok := cancelSet[pbresp.WatchId]; ok {
					break
				}

				cancelSet[pbresp.WatchId] = struct{}{}
				cr := &pb.WatchRequest_CancelRequest{
					CancelRequest: &pb.WatchCancelRequest{
						WatchId: pbresp.WatchId,
					},
				}
				req := &pb.WatchRequest{RequestUnion: cr}
				w.lg.Debug("sending watch cancel request for failed dispatch", zap.Int64("watch-id", pbresp.WatchId))
				if err := wc.Send(req); err != nil {
					w.lg.Debug("failed to send watch cancel request", zap.Int64("watch-id", pbresp.WatchId), zap.Error(err))
				}
			}

		// 监视客户端在Recv上失败；如果可能的话，生成另一个
		case err := <-w.errc:
			if isHaltErr(w.ctx, err) || toErr(w.ctx, err) == v3rpc.ErrNoLeader {
				closeErr = err
				return
			}
			if wc, closeErr = w.newWatchClient(); closeErr != nil {
				return
			}
			if ws := w.nextResume(); ws != nil {
				if err := wc.Send(ws.initReq.toPB()); err != nil {
					w.lg.Debug("error when sending request", zap.Error(err))
				}
			}
			cancelSet = make(map[int64]struct{})

		case <-w.ctx.Done():
			return

		case ws := <-w.closingc:
			w.closeSubstream(ws)
			delete(closing, ws)
			// 此流上不再有观察者，关闭，跳过取消
			if len(w.substreams)+len(w.resuming) == 0 {
				return
			}
			if ws.id != -1 {
				// 客户端正在关闭已建立的观察；在服务器上主动关闭它，而不是等到下一条消息到达时等待
				// 关闭
				cancelSet[ws.id] = struct{}{}
				cr := &pb.WatchRequest_CancelRequest{
					CancelRequest: &pb.WatchCancelRequest{
						WatchId: ws.id,
					},
				}
				req := &pb.WatchRequest{RequestUnion: cr}
				w.lg.Debug("sending watch cancel request for closed watcher", zap.Int64("watch-id", ws.id))
				if err := wc.Send(req); err != nil {
					w.lg.Debug("failed to send watch cancel request", zap.Int64("watch-id", ws.id), zap.Error(err))
				}
			}
		}
	}
}

// nextResume选择下一条恢复以注册grpc流。放弃的
// 由于头部必须等待其机上注册，因此队列中将流标记为nil。
func (w *watchGrpcStream) nextResume() *watcherStream {
	for len(w.resuming) != 0 {
		if w.resuming[0] != nil {
			return w.resuming[0]
		}
		w.resuming = w.resuming[1:len(w.resuming)]
	}
	return nil
}

// dispatchEvent将WatchResponse发送到相应的观察程序流
func (w *watchGrpcStream) dispatchEvent(pbresp *pb.WatchResponse) bool {
	events := make([]*Event, len(pbresp.Events))
	for i, ev := range pbresp.Events {
		events[i] = (*Event)(ev)
	}
	// TODO:返回观察ID？
	wr := &WatchResponse{
		Header:          *pbresp.Header,
		Events:          events,
		CompactRevision: pbresp.CompactRevision,
		Created:         pbresp.Created,
		Canceled:        pbresp.Canceled,
		cancelReason:    pbresp.CancelReason,
	}

	// 监视ID的索引为零，因此请求通知监视响应被分配一个-1的监视ID给
	// 以指示它们应该被广播。
	if wr.IsProgressNotify() && pbresp.WatchId == -1 {
		return w.broadcastResponse(wr)
	}

	return w.unicastResponse(wr, pbresp.WatchId)

}

// broadcastResponse向所有监视子流发送一个监视响应。
func (w *watchGrpcStream) broadcastResponse(wr *WatchResponse) bool {
	for _, ws := range w.substreams {
		select {
		case ws.recvc <- wr:
		case <-ws.donec:
		}
	}
	return true
}

// unicastResponse向特定的监视子流发送监视响应。
func (w *watchGrpcStream) unicastResponse(wr *WatchResponse, watchId int64) bool {
	ws, ok := w.substreams[watchId]
	if !ok {
		return false
	}
	select {
	case ws.recvc <- wr:
	case <-ws.donec:
		return false
	}
	return true
}

// serveWatchClient将来自grpc流的消息转发到运行（）
func (w *watchGrpcStream) serveWatchClient(wc pb.Watch_WatchClient) {
	for {
		resp, err := wc.Recv()
		if err != nil {
			select {
			case w.errc <- err:
			case <-w.donec:
			}
			return
		}
		select {
		case w.respc <- resp:
		case <-w.donec:
			return
		}
	}
}

// serveSubstream转发来自运行（）的监视响应对于订阅服务器
func (w *watchGrpcStream) serveSubstream(ws *watcherStream, resumec chan struct{}) {
	if ws.closing {
		panic("created substream goroutine but substream is closing")
	}

	// nextRev是预期的最小下一版本
	nextRev := ws.initReq.rev
	resuming := false
	defer func() {
		if !resuming {
			ws.closing = true
		}
		close(ws.donec)
		if !resuming {
			w.closingc <- ws
		}
		w.wg.Done()
	}()

	emptyWr := &WatchResponse{}
	for {
		curWr := emptyWr
		outc := ws.outc

		if len(ws.buf) > 0 {
			curWr = ws.buf[0]
		} else {
			outc = nil
		}
		select {
		case outc <- *curWr:
			if ws.buf[0].Err() != nil {
				return
			}
			ws.buf[0] = nil
			ws.buf = ws.buf[1:]
		case wr, ok := <-ws.recvc:
			if !ok {
				// 关闭子流
				return
			}

			if wr.Created {
				if ws.initReq.retc != nil {
					ws.initReq.retc <- ws.outc
					// 以防止下一次写入占用缓冲通道
					// 中的插槽并发布重复的创建事件
					ws.initReq.retc = nil

					// 仅在请求时发送第一次创建事件
					if ws.initReq.createdNotify {
						ws.outc <- *wr
					}
					// 返回手表频道后，当前版本
					// 手表必须在商店版本恢复。这是必要的
					// 以下情况才能按预期工作：
					// wch:=m1。手表（“a”）
					// m2。Put（“a”，“b”）
					// /<-wch 
					// 如果修订仅绑定在第一个观察到的事件上，
					// 如果在发布Put之前wch断开，则在提交后重新连接
					// 将错过Put。
					if ws.initReq.rev == 0 {
						nextRev = wr.Header.Revision
					}
				}
			} else {
				// 手表的当前进展；<=存储版本
				nextRev = wr.Header.Revision
			}

			if len(wr.Events) > 0 {
				nextRev = wr.Events[len(wr.Events)-1].Kv.ModRevision + 1
			}
			ws.initReq.rev = nextRev

			// 已在上面发送创建的事件，
			// 观察者不应发布重复事件
			if wr.Created {
				continue
			}

			// 如果缓冲区太大，则暂停通道
			ws.buf = append(ws.buf, wr)
		case <-w.ctx.Done():
			return
		case <-ws.initReq.ctx.Done():
			return
		case <-resumec:
			resuming = true
			return
		}
	}
	// 如果缺少id的事件
}

func (w *watchGrpcStream) newWatchClient() (pb.Watch_WatchClient, error) {
	// 将所有子流标记为恢复
	close(w.resumec)
	w.resumec = make(chan struct{})
	w.joinSubstreams()
	for _, ws := range w.substreams {
		ws.id = -1
		w.resuming = append(w.resuming, ws)
	}
	// 去掉零，如果任何
	var resuming []*watcherStream
	for _, ws := range w.resuming {
		if ws != nil {
			resuming = append(resuming, ws)
		}
	}
	w.resuming = resuming
	w.substreams = make(map[int64]*watcherStream)

	// 在接受观察者取消时连接到grpc流
	stopc := make(chan struct{})
	donec := w.waitCancelSubstreams(stopc)
	wc, err := w.openWatchClient()
	close(stopc)
	<-donec

	// 服务所有非关闭流，即使存在客户端错误
	// 以便拆卸路径可以按预期关闭流。
	for _, ws := range w.resuming {
		if ws.closing {
			continue
		}
		ws.donec = make(chan struct{})
		w.wg.Add(1)
		go w.serveSubstream(ws, w.resumec)
	}

	if err != nil {
		return nil, v3rpc.Error(err)
	}

	// 从新grpc流接收数据
	go w.serveWatchClient(wc)
	return wc, nil
}

func (w *watchGrpcStream) waitCancelSubstreams(stopc <-chan struct{}) <-chan struct{} {
	var wg sync.WaitGroup
	wg.Add(len(w.resuming))
	donec := make(chan struct{})
	for i := range w.resuming {
		go func(ws *watcherStream) {
			defer wg.Done()
			if ws.closing {
				if ws.initReq.ctx.Err() != nil && ws.outc != nil {
					close(ws.outc)
					ws.outc = nil
				}
				return
			}
			select {
			case <-ws.initReq.ctx.Done():
				// 关闭的ws将从恢复中删除
				ws.closing = true
				close(ws.outc)
				ws.outc = nil
				w.wg.Add(1)
				go func() {
					defer w.wg.Done()
					w.closingc <- ws
				}()
			case <-stopc:
			}
		}(w.resuming[i])
	}
	go func() {
		defer close(donec)
		wg.Wait()
	}()
	return donec
}

// joinSubstreams等待所有子流Goroutine完成。
func (w *watchGrpcStream) joinSubstreams() {
	for _, ws := range w.substreams {
		<-ws.donec
	}
	for _, ws := range w.resuming {
		if ws != nil {
			<-ws.donec
		}
	}
}

var maxBackoff = 100 * time.Millisecond

// openWatchClient会重试打开一个监视客户端，直到成功或停止。
// 在“ws==nil&&err==nil”的情况下手动重试
// TODO:remove FailFast=false 
func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error) {
	backoff := time.Millisecond
	for {
		select {
		case <-w.ctx.Done():
			if err == nil {
				return nil, w.ctx.Err()
			}
			return nil, err
		default:
		}
		if ws, err = w.remote.Watch(w.ctx, w.callOpts...); ws != nil && err == nil {
			break
		}
		if isHaltErr(w.ctx, err) {
			return nil, v3rpc.Error(err)
		}
		if isUnavailableErr(w.ctx, err) {
			// 重试，但回退
			if backoff < maxBackoff {
				// 25%回退因子
				backoff = backoff + backoff/4
				if backoff > maxBackoff {
					backoff = maxBackoff
				}
			}
			time.Sleep(backoff)
		}
	}
	return ws, nil
}

// toPB将内部监视请求结构转换为其protobuf WatchRequest结构。
func (wr *watchRequest) toPB() *pb.WatchRequest {
	req := &pb.WatchCreateRequest{
		StartRevision:  wr.rev,
		Key:            []byte(wr.key),
		RangeEnd:       []byte(wr.end),
		ProgressNotify: wr.progressNotify,
		Filters:        wr.filters,
		PrevKv:         wr.prevKV,
		Fragment:       wr.fragment,
	}
	cr := &pb.WatchRequest_CreateRequest{CreateRequest: req}
	return &pb.WatchRequest{RequestUnion: cr}
}

// toPB将内部进度请求结构转换为protobuf WatchRequest结构。
func (pr *progressRequest) toPB() *pb.WatchRequest {
	req := &pb.WatchProgressRequest{}
	cr := &pb.WatchRequest_ProgressRequest{ProgressRequest: req}
	return &pb.WatchRequest{RequestUnion: cr}
}

func streamKeyFromCtx(ctx context.Context) string {
	if md, ok := metadata.FromOutgoingContext(ctx); ok {
		return fmt.Sprintf("%+v", md)
	}
	return ""
}
